package com.yida.gwj.common.handler;

import java.util.HashMap;

import org.apache.commons.validator.ValidatorException;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import com.yida.common.core.exception.ServiceException;
import com.yida.gwj.common.domain.mq.MqMessageBody;
import com.yida.gwj.common.domain.mq.MqSendRequest;
import com.yida.gwj.common.domain.mq.MqSendResult;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;

/**
 * mq发送消息处理程序
 *
 * @author liangp
 * @date 2023/02/15
 */
@Component
public class MqSendMessageHandler {

    private final Logger log = LoggerFactory.getLogger(MqSendMessageHandler.class);

    private static final int MAX_DELAY_LEVEL = 20;

    @Autowired
    private RocketMQTemplate rocketMqTemplate;

    /**
     * 普通消息
     * <p>
     * level == 0，消息为非延迟消息1<=level<=maxLevel，消息延迟特定时间，例如level1，延迟1s level > maxLevel，则level maxLevel，例如level==20，延迟2h
     *
     * @param message
     * @param <T>
     * @return
     */
    public <T extends MqMessageBody, S extends MqSendRequest> MqSendResult sendMessage(T message, S request) {
        try {
            validator(request);
            HashMap<String, Object> keyMaps = new HashMap<String, Object>(11);
            keyMaps.put("KEYS", request.getKey());
            rocketMqTemplate.convertAndSend(request.getTopic() + ":" + request.getTag(), message, keyMaps);
            log.info("MQ消息[{}]发送成功", JSONUtil.toJsonStr(message));
            return new MqSendResult.Builder().success(true).message("发送MQ消息成功").build();
        } catch (Exception e) {
            log.error("MQ消息[" + JSONUtil.toJsonStr(message) + "]发送异常{}", e);
            return new MqSendResult.Builder().success(false).message("发送MQ消息失败 -> " + message).build();
        }
    }

    /**
     * 发送消息采用同步模式，这种方式只有在消息完全发送完成之后才返回结果，此方式存在需要同步等待发送结果的时间代价
     * <p>
     * level == 0，消息为非延迟消息1<=level<=maxLevel，消息延迟特定时间，例如level1，延迟1s level > maxLevel，则level maxLevel，例如level==20，延迟2h
     *
     * @param message
     * @param <T>
     * @return
     */
    public <T extends MqMessageBody, S extends MqSendRequest> MqSendResult syncSend(T message, S request) {
        try {
            validator(request);
            Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader("KEYS", request.getKey()).build();
            SendResult sendResult = rocketMqTemplate.syncSend(request.getTopic() + ":" + request.getTag(), sendMessage);
            if (ObjectUtil.notEqual(SendStatus.SEND_OK, sendResult.getSendStatus())) {
                log.error("MQ同步消息[{}]发送失败", JSONUtil.toJsonStr(message));
                return new MqSendResult.Builder().success(false).message("MQ同步消息发送失败 -> " + request.getKey()).build();
            }
            log.info("MQ同步消息[{}]发送成功", JSONUtil.toJsonStr(message));
            return new MqSendResult.Builder().success(true).message("MQ同步消息发送成功").build();
        } catch (Exception e) {
            log.error("MQ同步消息[" + JSONUtil.toJsonStr(message) + "]发送异常{}", e);
            return new MqSendResult.Builder().success(false).message("MQ同步消息发送失败 -> " + request.getKey()).build();
        }
    }

    /**
     * 发送消息采用异步发送模式，消息发送后立刻返回，当消息完全完成发送后，会调用回调函数sendCallback来告知发送者本次发送是成功或者失败。
     * <p>
     * level == 0，消息为非延迟消息1<=level<=maxLevel，消息延迟特定时间，例如level1，延迟1s level > maxLevel，则level maxLevel，例如level==20，延迟2h
     *
     * @param message
     * @param <T>
     * @return
     */
    public <T extends MqMessageBody, S extends MqSendRequest> MqSendResult asyncSendMessage(T message, S request) {
        try {
            validator(request);
            Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader("KEYS", request.getKey()).build();
            MqSendResult mqSendResult = new MqSendResult.Builder().build();
            rocketMqTemplate.asyncSend(request.getTopic() + ":" + request.getTag(), sendMessage, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("MQ异步消息[{}]发送成功", JSONUtil.toJsonStr(message));
                    mqSendResult.setSuccess(true);
                    mqSendResult.setMessage("MQ异步消息发送成功");
                }

                @Override
                public void onException(Throwable throwable) {
                    log.error("MQ异步消息[{}]发送失败", JSONUtil.toJsonStr(message));
                    mqSendResult.setSuccess(false);
                    mqSendResult.setMessage("MQ异步消息发送失败 -> " + message);
                    throw new ServiceException("MQ异步消息发送失败," + throwable);
                }
            });
            return mqSendResult;
        } catch (Exception e) {
            log.error("MQ消息[" + JSONUtil.toJsonStr(message) + "]发送异常{}", e);
            return new MqSendResult.Builder().success(false).message("MQ异步消息发送失败 -> " + request.getKey()).build();
        }
    }

    /**
     * 采用one-way发送模式发送消息的时候，发送端发送完消息后会立即返回，不会等待来自broker的ack来告知本次消息发送是否完全完成发送
     * <p>
     * level == 0，消息为非延迟消息1<=level<=maxLevel，消息延迟特定时间，例如level1，延迟1s level > maxLevel，则level maxLevel，例如level==20，延迟2h
     *
     * @param message
     * @param <T>
     * @return
     */
    public <T extends MqMessageBody, S extends MqSendRequest> MqSendResult oneWaySend(T message, S request) {
        try {
            validator(request);
            Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader("KEYS", request.getKey()).build();
            rocketMqTemplate.sendOneWay(request.getTopic() + ":" + request.getTag(), sendMessage);
            log.info("MQ消息[{}]发送成功", JSONUtil.toJsonStr(message));
            return new MqSendResult.Builder().success(true).message("发送MQ消息成功").build();
        } catch (Exception e) {
            log.error("MQ消息[" + JSONUtil.toJsonStr(message) + "]发送异常{}", e);
            return new MqSendResult.Builder().success(false).message("发送MQ消息失败 -> " + request.getKey()).build();
        }
    }

    /**
     * 发送半消息采用事务模式，这种方式只有在消息完全发送完成之后才返回结果，但是只有TransactionListener返回成功才会进行消费
     * <p>
     * TransactionListener状态: 消费提交: COMMIT_MESSAGE, 消费回滚: ROLLBACK_MESSAGE, 异常重新监听事务: UNKNOW;
     * <p>
     * level == 0，消息为非延迟消息1<=level<=maxLevel，消息延迟特定时间，例如level1，延迟1s level > maxLevel，则level maxLevel，例如level==20，延迟2h
     *
     * @param message
     * @param <T>
     * @return
     */
    public <T extends MqMessageBody, S extends MqSendRequest, K extends TransactionListener> MqSendResult sendMessageInTransaction(T message, S request, Object txMessage, K halfListener) {
        try {
            validator(request);
            // rocketMqTemplate.getProducer().setSendMsgTimeout(request.getLevel().getCode());
            ((TransactionMQProducer)rocketMqTemplate.getProducer()).setTransactionListener(halfListener);
            Message<T> sendMessage = MessageBuilder.withPayload(message).setHeader("KEYS", request.getKey()).build();
            SendResult sendResult = rocketMqTemplate.sendMessageInTransaction(request.getTopic() + ":" + request.getTag(), sendMessage, txMessage);
            if (ObjectUtil.notEqual(SendStatus.SEND_OK, sendResult.getSendStatus())) {
                log.error("MQ事务消息[{}]发送失败", JSONUtil.toJsonStr(message));
                return new MqSendResult.Builder().success(false).message("MQ事务消息发送失败 -> " + request.getKey()).build();
            }
            log.info("MQ事务消息[{}]发送成功", JSONUtil.toJsonStr(message));
            return new MqSendResult.Builder().success(true).message("MQ事务消息发送发送成功").build();
        } catch (Exception e) {
            log.error("MQ事务消息[" + JSONUtil.toJsonStr(message) + "]发送异常{}", e);
            return new MqSendResult.Builder().success(false).message("MQ事务消息发送发送失败 -> " + request.getKey()).build();
        }
    }

    /**
     * mq请求参数校验
     *
     * @param request 请求
     */
    private void validator(MqSendRequest request) throws ValidatorException {
        if (StrUtil.isBlank(request.getTag())) {
            log.error("tag不能为空");
            throw new ValidatorException("tag不能为空");
        }
        if (StrUtil.isBlank(request.getTopic())) {
            log.error("topic不能为空");
            throw new ValidatorException("topic不能为空");
        }
        if (StrUtil.isBlank(request.getKey())) {
            log.error("key不能为空");
            throw new ValidatorException("key不能为空");
        }
        if (!(request.getLevel().getCode() >= 0 && request.getLevel().getCode() <= MAX_DELAY_LEVEL)) {
            log.error("level设置错误");
            throw new ValidatorException("level设置错误");
        }
    }

}
